package load

import (
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/golang/snappy"
	"hermes/manager"
	"log"
	"os"
	"time"
)

var sender sarama.SyncProducer

func InitLoader(){
	//TODO need kafka init config
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 5 * time.Second
	p, err := sarama.NewSyncProducer(manager.GetLoadConfig().BrokerList, config)
	if err != nil {
		log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
		return
	}
	sender = p
	//defer sender.Close()
}

func KafkaSend(topic string,content []byte)  {
	if sender == nil {
		InitLoader()
	}
	isSnappy:=false
	var contentCompress []byte
	if isSnappy{
		//snappy
		contentCompress=snappy.Encode(nil,  content)
	}else{
		contentCompress = content
	}

	msg := &sarama.ProducerMessage{
		Topic:topic,
		Value:sarama.ByteEncoder(contentCompress),
	}
	part, offset, err := sender.SendMessage(msg)
	if err != nil {
		log.Printf("send message(%s) to (%s) err=%s \n", content,topic, err)
	}else {
		fmt.Fprintf(os.Stdout, string(content) + "发送成功，topic=%s partition=%d, offset=%d \n",topic, part, offset)
	}
}

